package com.yxc.emqxdemo.configuration;

import com.yxc.emqxdemo.utils.SpringContextHolder;
import com.yxc.emqxdemo.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * emqx收到消息的回调
 *
 * @author: YXC
 * @time: 2022/9/15 11:21
 **/
@Slf4j
@Component
public class EmqxCallback implements MqttCallback {

    Map<String, EmqxConsumerParent> consumerBean = new HashMap<>();
    @Autowired
    private EmqxClient emqxClient;
    @Autowired
    @Qualifier("emqxExecutor")
    private ThreadPoolTaskExecutor emqxExecutor;

    public EmqxCallback() throws ClassNotFoundException, MqttException {
        for (String beanName : SpringContextHolder.getApplicationContext().getBeanDefinitionNames()) {
            try {
                Object bean = SpringContextHolder.getBean(beanName);
                EmqxConsumer anno = bean.getClass().getAnnotation(EmqxConsumer.class);
                if (anno != null) {
                    consumerBean.put(anno.topic(), (EmqxConsumerParent) SpringContextHolder.getBean(beanName));
                }
            } catch (Exception e) {
                log.error("bean({})未获取", beanName);
            }
        }
    }

    @Bean(name = "emqxExecutor")
    public ThreadPoolTaskExecutor emqxExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 设置核心线程数
        executor.setCorePoolSize(5);
        // 设置最大线程数
        executor.setMaxPoolSize(50);
        // 设置队列容量
        executor.setQueueCapacity(100);
        // 设置线程活跃时间（秒）
        executor.setKeepAliveSeconds(30);
        // 设置默认线程名称
        executor.setThreadNamePrefix("emqx-");
        // 设置拒绝策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 等待所有任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        return executor;
    }

    /**
     * 连接断开
     *
     * @author YXC
     * @date 2022/9/15 11:22
     **/
    @Override
    public void connectionLost(Throwable cause) {
        cause.printStackTrace();
        log.error("emqx连接中断: {}", cause.getMessage());
        try {
            emqxClient.reconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 收到消息
     *
     * @param topic:   接收消息的topic
     * @param message: 接收消息的内容
     * @author YXC
     * @date 2022/9/15 11:23
     **/
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        try {
            String payload = new String(message.getPayload());
            log.debug("收到[{}]的(qos:{})消息: {}", topic, message.getQos(), payload);
            if (consumerBean.containsKey(topic)) {
                EmqxConsumerParent consumer = consumerBean.get(topic);
                EmqxConsumer anno = consumer.getClass().getAnnotation(EmqxConsumer.class);
                if (anno.mode().equals(EmqxConsumer.Mode.SINGLE)) {
                    // 单例消费
                    emqxExecutor.execute(() -> {
                        consumerBean.get(topic).receiveMessage(payload);
                    });
                } else if (EmqxConsumer.Mode.BATH.equals(anno.mode())) {
                    // 批量消费
                }
                // consumerBean.get(topic).receiveMessage(payload);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 消息接收成功
     *
     * @param token:
     * @author YXC
     * @date 2022/9/15 11:25
     **/
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            MqttMessage message = token.getMessage();
            if (message != null) {
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                String[] topics = token.getTopics();
                log.debug("topic: [{}] 成功接收消息: {} ", StringUtils.arrayToString(topics, ","), payload);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public Map<String, EmqxConsumerParent> getConsumerBean() {
        return this.consumerBean;
    }
}
